package com.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import javax.swing.*;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class DMLApi {

    private Configuration conf;
    private Connection conn;
    private Admin admin;

    @Before
    public void init() throws IOException {
        // 创建配置文件
        conf = HBaseConfiguration.create();
        // 加载zookeeper的配置
        conf.set("hbase.zookeeper.quorum", "node02,node03,node04");
        // 获取连接
        conn = ConnectionFactory.createConnection(conf);
        // 获取admin对象
        admin = conn.getAdmin();
    }

    @Test
    public void hbaseSplit() throws IOException {
        //自定义算法，产生一系列Hash散列值存储在二维数组中
        byte[][] splitKeys = {{'a', 'b'}, {'d', 'e'}};
        // 构建 Table 描述
        TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf("t_split5"));
        // 给table添加列族
        ColumnFamilyDescriptorBuilder columnBuilder =
                ColumnFamilyDescriptorBuilder.newBuilder("info".getBytes());

        tableBuilder.setColumnFamily(columnBuilder.build());

        admin.createTable(tableBuilder.build(), splitKeys);
    }


    public static void main(String[] args) {
        String str = "1500100007";
        System.out.println(str.matches("15001000\\w+"));
    }


    /**
     * get读取一行数据
     */
    @Test
    public void getTest() throws Exception {
        // 要先获取表
        TableName tableName = TableName.valueOf("default", "HBASE_STUDENT3");
        // 获取table对象
        Table table = conn.getTable(tableName);

        // 创建get对象
        Get get = new Get("stu_1500100992".getBytes());

        // 获取查询的结果
        Result result = table.get(get);

        String id = new String(result.getRow());
        System.out.println(id);
        String name = new String(result.getValue("CF1".getBytes(), "name".getBytes()));
        String age = new String(result.getValue("CF1".getBytes(), "age".getBytes()));
        String gender = new String(result.getValue("CF1".getBytes(), "gender".getBytes()));
        String clazz = new String(result.getValue("CF2".getBytes(), "clazz".getBytes()));
        System.out.println("name:" + name + ",age:" + age + ",gender:" + gender + ",clazz:" + clazz);

        // 解析result，获取列的值
//        Cell[] cells = result.rawCells();
//
//        for (Cell cell : cells) {
//            byte[] rowkeyBys = CellUtil.cloneValue(cell);
//            System.out.println(new String(rowkeyBys));
//        }

        table.close();

    }


    /**
     * 删除数据
     */
    @Test
    public void deleteTest() throws Exception {
        // 要先获取表
        TableName tableName = TableName.valueOf("api_test", "student");
        // 获取table对象
        Table table = conn.getTable(tableName);

        // 指定删除的rowkey，把一整行数据都删除了
        Delete delete = new Delete("1500100004".getBytes());

        // 控制只删除某一列
        delete.addColumn("info".getBytes(), "name".getBytes());

        // addColumns, 可以删除多个版本

        // 删除指定rowkey的数据
        table.delete(delete);

        table.close();
    }

    /**
     * 批量put一批数据到HBase
     */
    @Test
    public void putAllTest() throws Exception {
        // 要先获取表
        TableName tableName = TableName.valueOf("api_test", "student");
        // 获取table对象
        Table table = conn.getTable(tableName);

        // 创建put 集合
        List<Put> putList = new ArrayList<>();

        BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));

        String line;
        while ((line = br.readLine()) != null) {
            String[] split = line.split(",");
            // rowkey,info:name,info:age, info:gender, info:clazz
            Put put = new Put(split[0].getBytes());

            put.addColumn("info".getBytes(), "name".getBytes(), split[1].getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), split[2].getBytes());
            put.addColumn("info".getBytes(), "gender".getBytes(), split[3].getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), split[4].getBytes());

            putList.add(put);
        }
        // 批量保存put数据
        table.put(putList);

        table.close();
    }


    /**
     * put一条数据到hbase表，添加、修改
     */
    @Test
    public void putTest() throws Exception {
        // 要先获取表
        TableName tableName = TableName.valueOf("api_test", "student");

        // 获取table对象
        Table table = conn.getTable(tableName);

        // 创建put对象
        // put 'table', 'rowkey', 'info:name', 'zs'
        Put put = new Put("1500100004".getBytes());
        put.addColumn("info".getBytes(), "name".getBytes(), Bytes.toBytes("晓明"));

        // 添加数据
        table.put(put);

        table.close();
    }

    @Test
    public void test1() {
        System.out.println(conn);
        System.out.println(admin);
    }


    @After
    public void afterTest() throws IOException {
        admin.close();
        conn.close();
    }
}
